package com.scala.business

import com.google.gson.Gson
import org.apache.spark.rdd.RDD
import com.scala.pojo.AllBasicInfoScala
import org.apache.spark.{SparkConf, SparkContext}

/**
 * 1、计算乘机次数和总飞行公里数的比值,并平均飞行距离前20的会员
 * - 计算每个会员的乘机次数和总飞行公里数的比值，来了解每次乘机的平均飞行距离。
 * - 考虑到乘机次数和航班距离可能会对票价产生影响，该指标可以帮助我们更深入的了解客户的飞行习惯。
 *
 * val 和 var 的区别：
 * val:声明的变量是不可变的（immutable），即一旦被赋值后就不能再改变。
 * var:声明的变量是可变的（mutable），即其值可以多次改变
 */
object Take01 {
  val gson: Gson = new Gson()

  def main(args: Array[String]): Unit = {
    val sparkConf: SparkConf = new SparkConf().setAppName("平均飞行距离前20的会员").setMaster("local")
    val sc: SparkContext = new SparkContext(sparkConf)
    val rdd: RDD[String] = sc.textFile("hdfs://master:9000/air_data/*")

    val take = rdd.map(x => {
        val gson: Gson = new Gson()
        val aLlBasicInfo: AllBasicInfoScala = gson.fromJson(x, classOf[AllBasicInfoScala])

        val flightCount: Int = aLlBasicInfo.flightCount
        val segKmSum: Long = aLlBasicInfo.segKmSum

        // 计算每个会员的平均飞行距离
        var avgFlight: Long = 0L
        try {
          avgFlight = segKmSum / flightCount
        } catch { //case _: Exception => 表示捕获任意类型的 Exception 异常
          case _: Exception =>
        }

        (avgFlight, aLlBasicInfo.memberNo)
      }).groupByKey()
      .sortByKey(ascending = false)
      .take(20)

    println("平均飞行距离Top20的会员:")
    var i: Int = 0 // 记录排名
    for (avgFlight <- take) {
      i += 1
      printf("top%d -->平均飞行距离为 %s 的会员有：%s", i, avgFlight._1, avgFlight._2)
      println()
    }
    /*CompactBuffer 是 Scala 集合框架中的一个类，它实现了可变序列 Buffer 的接口。CompactBuffer 提供了一个高效的存储和访问元素的方式。
    * CompactBuffer 使用数组内部进行元素的存储，因此可以提供快速的随机访问和更新操作*/

    sc.stop()

  }
}
